package com.jh.paymentchannel.repository;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Range;
import com.jh.paymentchannel.payment.TradeStatus;
import com.jh.paymentchannel.pojo.PaymentOrder;
import com.jh.paymentchannel.pojo.UserTurnover;
import org.jetbrains.annotations.Nullable;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.data.jpa.repository.support.SimpleJpaRepository;
import org.springframework.transaction.annotation.Transactional;

import javax.persistence.EntityManager;
import javax.persistence.TypedQuery;
import javax.persistence.criteria.CriteriaBuilder;
import javax.persistence.criteria.CriteriaQuery;
import javax.persistence.criteria.Path;
import javax.persistence.criteria.Predicate;
import javax.persistence.criteria.Root;
import javax.persistence.criteria.Selection;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import static com.google.common.base.Preconditions.checkArgument;


/**
 * @author Fuchun
 * @since 1.0
 */
public class PaymentOrderRepositoryImpl extends SimpleJpaRepository<PaymentOrder, Long>
        implements PaymentOrderRepositoryCustom {

    private final EntityManager em;

    public PaymentOrderRepositoryImpl(EntityManager em) {
        super(PaymentOrder.class, em);
        this.em = em;
    }

    private void checkDateRange(Range<Date> dateRange) {
        checkArgument(dateRange != null, "The dateRange must be not null.");
        checkArgument(dateRange.hasLowerBound() && dateRange.hasUpperBound(),
                "The dateRange must have lower and upper endpoint.");
    }

    @Override
    @Transactional(readOnly = true)
    public List<Long> findUserIdsOnTradeSuccess(Range<Date> dateRange) {
        checkDateRange(dateRange);
        Date from = dateRange.lowerEndpoint();
        Date to = dateRange.upperEndpoint();

        CriteriaBuilder cb = em.getCriteriaBuilder();
        CriteriaQuery<Long> userIdQuery = cb.createQuery(Long.class);
        Path<PaymentOrder> root = userIdQuery.from(PaymentOrder.class);

        Path<Long> userId = root.get("userId");
        Predicate betweenAnd = cb.between(root.get("createTime"), from, to);
        Predicate tradeSuccess = cb.equal(root.get("tradeStatus"), TradeStatus.TRADE_SUCCESS);
        Predicate userIdNotNull = cb.isNotNull(userId);

        userIdQuery.where(betweenAnd, tradeSuccess, userIdNotNull)
                .select(userId).orderBy(cb.asc(userId));
        TypedQuery<Long> typedQuery = em.createQuery(userIdQuery);
        return typedQuery.getResultList();
    }

    @Override
    public Map<Long, UserTurnover> findUserTurnovers(
            Range<Date> dateRange, @Nullable Specification<PaymentOrder> otherSpec) {
        checkDateRange(dateRange);
        Date from = dateRange.lowerEndpoint();
        Date to = dateRange.upperEndpoint();

        CriteriaBuilder cb = em.getCriteriaBuilder();
        CriteriaQuery<UserTurnover> criteriaQuery = cb.createQuery(UserTurnover.class);
        Root<PaymentOrder> root = criteriaQuery.from(PaymentOrder.class);

        String uidAlias = "userId", coIdAlias = "countOrderId", stfAlias = "sumTotalFee";

        // SELECT po.userId,
        Path<Long> userId = root.get(uidAlias);
        // COUNT(po.orderId) as `countOrderId`,
        Selection<Integer> countOrderId = cb.count(root.get("orderId")).as(Integer.class).alias(coIdAlias);
        // SUM(po.totalFee) as `sumTotalFee`,
        Selection<BigDecimal> sumTotalFee = cb.sum(root.get("totalFee")).as(BigDecimal.class).alias(stfAlias);
        List<Predicate> predicates = new ArrayList<>(4);
        // WHERE po.createTime BETWEEN {from} AND {to}
        predicates.add(cb.between(root.get("createTime"), from, to));
        // AND po.tradeStatus = 'TRADE_SUCCESS'
        predicates.add(cb.equal(root.get("tradeStatus"), TradeStatus.TRADE_SUCCESS));
        // AND po.userId IS NOT NULL
        predicates.add(cb.isNotNull(userId));
        if (otherSpec != null) {
            predicates.add(otherSpec.toPredicate(root, criteriaQuery, cb));
        }
        // GROUP BY po.userId ORDER BY po.userId ASC
        criteriaQuery.where(predicates.toArray(new Predicate[0]))
                .select(cb.construct(UserTurnover.class, userId, countOrderId, sumTotalFee))
                .groupBy(userId).orderBy(cb.asc(userId));
        TypedQuery<UserTurnover> query = em.createQuery(criteriaQuery);
        List<UserTurnover> turnovers = query.getResultList();
        Map<Long, UserTurnover> resultMap = ImmutableMap.of();
        if (turnovers != null && !turnovers.isEmpty()) {
            resultMap = new LinkedHashMap<>(turnovers.size());
            for (UserTurnover turnover : turnovers) {
                resultMap.put(turnover.getUserId(), turnover);
            }
        }
        return resultMap;
    }

    @Override
    public Page<PaymentOrder> findTradingSuccess(
            Range<Date> dateRange, @Nullable Specification<PaymentOrder> otherSpec, Pageable pageable) {
        checkDateRange(dateRange);
        Date from = dateRange.lowerEndpoint();
        Date to = dateRange.upperEndpoint();
        Specification<PaymentOrder> spec = (root, query, cb) -> {
            List<Predicate> predicates = new ArrayList<>(4);
            // WHERE po.createTime BETWEEN {from} AND {to}
            predicates.add(cb.between(root.get("createTime"), from, to));
            // AND po.tradeStatus = 'TRADE_SUCCESS'
            predicates.add(cb.equal(root.get("tradeStatus"), TradeStatus.TRADE_SUCCESS));
            // AND po.userId IS NOT NULL
            predicates.add(cb.isNotNull(root.get("userId")));
            if (otherSpec != null) {
                predicates.add(otherSpec.toPredicate(root, query, cb));
            }
            return cb.and(predicates.toArray(new Predicate[0]));
        };
        return findAll(spec, pageable);
    }
}
